package broadcast;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.util.HashMap;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Queue;

import common.Entity;
import common.Message;


class ReceiveMgr extends Thread{
		private String componentId;
		private Broadcast parent;
		private DatagramSocket listener;
		private DatagramPacket p;
		private Map<Entity, Integer> entityRecSeq;
		private Map<Entity, Queue<Message>> messages;
		private MessageLog log;
		
		public ReceiveMgr(Broadcast b) {
			parent = b;
			messages = new HashMap<Entity, Queue<Message>>();
			entityRecSeq = new HashMap<Entity, Integer>();
			log = new MessageLog();
			try {
				listener = new DatagramSocket(b.getListenPort());
			} catch (SocketException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			byte[] buffer = new byte[Serializer.MAXLENGTH * 2]; 
			p = new DatagramPacket(buffer, Serializer.MAXLENGTH * 2);
		}
		
		public void run(){
			while(true) {
				try {
					listener.receive(p);
					int length = p.getLength();
					byte[] data = new byte[p.getLength()];
					System.arraycopy(p.getData(), p.getOffset(), data, 0, length);
					Message m = parent.getSerializer().merge(data);
					if(m != null) {
						if(m.getDest() == null) {
//							System.out.println("something wrong");
						}
						String s = new String(m.getData());
						tryDeliver(m);
					}
				} catch (SocketException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			} 
		}
		
		private void tryDeliver(Message	m) {
			Entity src = m.getSource();
//			if(m.getViewNum() == Broadcast.NEW_VIEW){
//				if(log.contains(m)) {
//					System.out.println("discard");
//					return;
//				} else {
//					log.addMessage(m);
//				}
//				dumpQueue(src);
//				entityRecSeq.put(src, m.getSeq());
//				tryAddtoQueue(m);
//				dumpQueue(src);
//				return;
//			}
			if(!entityRecSeq.containsKey(src)) {
				entityRecSeq.put(src, m.getSeq());
				tryAddtoQueue(m);
				dumpQueue(src);
			} else {
				if(m.getSeq() >= entityRecSeq.get(src)){
					tryAddtoQueue(m);
					dumpQueue(src);
				}
			}
		}
		
		private void tryAddtoQueue(Message m) {
			Entity src = m.getSource();
			Queue<Message> msgs;
			if(!messages.containsKey(src)) {
				msgs = new PriorityQueue<Message>();
				messages.put(src, msgs);
			} else {
				msgs = messages.get(src);
			}
			msgs.add(m);
		}
		
//		private void emptyQueue(Entity src) {
//			messages.remove(src);
//		}
		
		private void dumpQueue(Entity src) {
			Queue<Message> q = messages.get(src);
			if(q == null){
				return;
			}
			while(!q.isEmpty()) {
				Message m = q.peek();
				if(m.getSeq() == entityRecSeq.get(src)) {
					q.remove();
					entityRecSeq.put(src, m.getSeq() + 1);
					parent.sendMsg(m);
					parent.deliver(m);
				} else {
					break;
				}
			}
		}
		
		void resetEntity(Entity key) {
			entityRecSeq.remove(key);
			messages.remove(key);
		}
		
		synchronized void setComponentId(String s){
			componentId = s;
		}
}